package com.bsc.rtc.lib.libraries.RxJavaRetrofitDemo;

import android.annotation.SuppressLint;
import android.os.Bundle;
import android.util.Log;
import android.view.View;

import com.bsc.rtc.lib.libraries.R;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

import androidx.appcompat.app.AppCompatActivity;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.ObservableTransformer;
import io.reactivex.Observer;
import io.reactivex.Scheduler;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.schedulers.Schedulers;


public class RxDemoActivity extends AppCompatActivity {
    private String TAG = this.getClass().getSimpleName();
    Observer observer;

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_rx_demo);
        initObserver();
    }

    public void onClickView(View v) {
        switch (v.getId()) {
            case R.id.test:
                // Test();
                //testFromIterable();
                //testTimer();
                // testInteval();
                // testIntevalRange();
                //testRange();
                // testMap();
                //   testFlatMap();
                //testContact();
                //testMerge();
                // testDoOperate();
                testCompos();
                break;

        }
    }

    private void initObserver() {
        observer = new Observer() {
            Disposable disposable;

            @Override
            public void onSubscribe(Disposable d) {
                Log.i(TAG, "onNext start observer");
                disposable = d;
            }

            @Override
            public void onNext(Object o) {
                Log.i(TAG, "onNext " + o + "\n");
            }

            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "error" + e.getMessage());
            }

            @Override
            public void onComplete() {
                Log.i(TAG, "onComplete");
            }
        };
    }

    private void Test() {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                Log.i(TAG, "Observable emit 1" + "\n");
                emitter.onNext(1);
                Log.i(TAG, "Observable emit 2" + "\n");
                emitter.onNext(2);
                Log.i(TAG, "Observable emit 3" + "\n");
                emitter.onNext(3);
                emitter.onComplete();
                Log.i(TAG, "Observable emit 4" + "\n");
                emitter.onNext(4);

            }
        }).subscribe(observer);

    }

    private void testFromIterable() {
        List<Integer> list = new ArrayList() {
        };
        list.add(3);
        list.add(4);
        list.add(5);
        list.add(6);
        Observable.fromIterable(list).subscribe(observer);
    }

    private void testTimer() {
        Observable.timer(1, TimeUnit.SECONDS, new Scheduler() {

            @Override
            public Worker createWorker() {
                return null;
            }
        }).subscribe(observer);
    }

    private void testInteval() {
        Observable.interval(2, 1, TimeUnit.SECONDS).subscribe(observer);
    }

    private void testIntevalRange() {
        Observable.intervalRange(3, 10, 2, 1, TimeUnit.SECONDS).subscribe(observer);
    }

    private void testRange() {
        Observable.range(4, 10).subscribe(observer);
    }

    private void testMap() {
        List<Integer> list = new ArrayList() {
        };
        list.add(3);
        list.add(4);
        list.add(5);
        list.add(6);
        Observable.fromIterable(list).map(integer -> "toString" + integer).subscribe(observer);

    }

    private void testFlatMap() {
        List<Integer> list = new ArrayList() {
        };
        list.add(3);
        list.add(4);
        list.add(5);
        list.add(6);
        Observable.fromIterable(list).flatMap(new Function<Integer, ObservableSource<String>>() {
            @Override
            public ObservableSource<String> apply(Integer integer) throws Exception {
                final List<String> list1 = new ArrayList<>();
                for (int i = 0; i < 3; i++) {
                    list1.add("我是事件 " + integer + "拆分后的子事件" + i);
                    // 通过flatMap中将被观察者生产的事件序列先进行拆分，再将每个事件转换为一个新的发送三个String事件
                    // 最终合并，再发送给被观察者
                }
                return Observable.fromIterable(list1);
            }
        }).subscribe(observer);
    }

    private void testContact() {
        Observable.concat(Observable.just(1, 2, 3, 4), Observable.just(5, 6), Observable.just(1, 2, 8, 9, 10)).subscribe(observer);
    }

    private void testMerge() {
        Observable.merge(Observable.intervalRange(0, 3, 1, 1, TimeUnit.SECONDS), // 从0开始发送、共发送3个数据、第1次事件延迟发送时间 = 1s、间隔时间 = 1s
                Observable.intervalRange(2, 3, 1, 1, TimeUnit.SECONDS)).subscribe(observer);
    }

    private void testDoOperate() {
        List<Integer> list = new ArrayList() {
        };
        list.add(3);
        list.add(4);
        list.add(5);
        list.add(6);
        Observable.fromIterable(list).doOnEach(new Observer() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(Object o) {
                Log.i(TAG, "do--- " + o);
            }

            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "do--- " + e.getMessage());
            }

            @Override
            public void onComplete() {
                Log.i(TAG, "do--- onComplete");
            }
        }).subscribe(observer);
    }

    @SuppressLint("CheckResult")
    private void testCompos() {
        Observable.create((ObservableOnSubscribe<String>) emitter -> {
            emitter.onNext("1");
            emitter.onNext("2");
            emitter.onNext("3");
            emitter.onNext("4");
        }).subscribeOn(AndroidSchedulers.mainThread())
                .observeOn(Schedulers.io())
                .compose(String2Int())
                .subscribe(s -> {

                    Log.i("zzw", "s==" + s);
                });
    }

    ObservableTransformer<String, Integer> String2Int() {
        return new ObservableTransformer<String, Integer>() {
            @Override
            public ObservableSource<Integer> apply(Observable<String> upstream) {
                return upstream.map(Integer::valueOf);
            }
        };
    }
}
